package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/2/5
 *
 *      tableEnvironment.executeSql(写操作 insert sql)
 *
 *      tableEnvironment.sqlQuery(读操作 select sql)
 */
public class Demo1_Sql
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());
        Table table = tableEnvironment.fromDataStream(ds);

        //使用SQL查询表
        //Table table1 = tableEnvironment.sqlQuery("select * from " + table + " where id = 's1'");

        //对目标表起个名字
        tableEnvironment.createTemporaryView("t1",table);
        //纯sql
        Table table1 = tableEnvironment.sqlQuery("select * from t1 where id = 's1'");

        table1.execute().print();


    }
}
